package cn.spark.study.core;

import java.util.Arrays;
import java.util.Iterator;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

/**
 * 将 JAVA 开发的 Word Count 程序，部署到集群
 * @author wdy
 *
 */
public class WordCountCluster {

    public static void main( String[] args )
    {
        System.out.println( "Hello WordCountCluster!" );
        
        // 第一步：创建 SparkConf 对象，设置 Spark 应用的配置信息
        // 使用 setMaster 可以设置 Spark 应用要连接的 Spark 集群的 Master 节点的 url
        // 但是如果设置成 local 则表示在本地运行
        
        // 集群方式需要修改
        // 1. setMaster 方法删掉，会默认寻找
        // 2. 读取的文件要设置成 hadoop hdfs 上的真正大数据文件
        
        // 实际执行步骤：
        // 1. 将 spark.txt 上传到 hdfs 上
        // hdfs 上传使用 hdfs fs -put spark.txt /spark.txt
        // 2. 打包工程
        // maven 打包插件
        // 3. 将 jar 包上传 spark1 服务器
        // 4. 编写 spark 集群执行脚本
        // 5. 执行脚本
//        /usr/local/spark/bin/spark-submit \
//        --class cn.spark.study.core.WordCountCluster \
//        --num-executors 3 \
//        --executor-cores 3 \
//        /mnt/spark-study/java/spark-study-java-1.6.0.jar
        SparkConf conf = new SparkConf()
        		.setAppName("WordCountCluster");
        		// .setMaster("local");
        
        // 第二步：创建 JavaSparkContext 对象
        // 在 Spark 中 SparkContext 是 Spark 所有功能的一个入口，无论是 java，scala，甚至是 python 编写，都需要一个 SparkContext，
        // 它的主要作用包括：初始化 Spark 应用程序所需要的一切核心组件，如：调度器（DAGSchedule, TaskSchedule）,还会到 Spark 节点上进行注册等等
        // 编写不同的 Spark 程序，使用的 SparkContet 是不同的。如果是 scala 是使用原生的 SparkContext
        // JAVA 是 JavaSparkContext
        // Spark SQL 是 SQLContext\HiveContext
        // Spark Streaming 是独有的 SparkContext 等等
        JavaSparkContext sc = new JavaSparkContext(conf);
        
        // 第三步：针对输入源（hdfs文件、本地文件等等），创建一个初始的 RDD
        // 输入源中的数据会打散，分配到 RDD 的每个 partition 中，从而形成一个初始的分布式的数据集
        // 这里用本地文件做测试
        // SparkContext 中用于根据文件类型的输入源创建 RDD 的方法，叫做 textFile() 方法
        // Java 中普通 RDD 都叫做 JavaRDD
        // RDD 中有元素的概念，如果是 hdfs或者本地文件，创建的 RDD 每一个元素都相当于是文件里的一行
        JavaRDD<String> lines = sc.textFile("hdfs://spark1:9000/spark.txt");
        
        // 第四步：对初始 RDD 进行 transformation 操作，也就是一些计算操作
        // 通常操作会通过创建 function，并配合 RDD 的 map、flatMap 等算子来执行 function
        // 通常比较简单的则创建指定 function的匿名内部类，但如果 function 比较复杂，
        // 则会单独创建一个类，作为实现这个 function 接口的类
        
        // 先将每一行拆分成单个的词
        // FlatMapFunction 两个参数，分别代表着输入和输出类型
        // 我们这里输入肯定是 String，因为是一行一行的文本。输出也是String，因为是每一行的文本
        // flatMap 算子的作用是把 RDD 的一个元素，拆分成一个或多个元素
        // JavaRDD 的两个泛型参数，分别代表
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

        	
			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			public Iterator<String> call(String line) throws Exception {
				return Arrays.asList(line.split(" ")).iterator();
			}

        });

        
        // 将每一个单词映射为 （单词,1）的形式
        // 这样可以进行单词出现次数的累加
        // mapToPair 是将每个元素英社称 （v1，v2）这样的 Tuple2 类型的元素
        // Tuple2 就是 scala 里面的 tuple 类型，包含两个值
        // mapToPair 算子要求是与 PairFunction 配合使用第一个泛型参数代表了输入类型，第二个、第三个泛型参数，代表输出的 Tuple2 的第一个、第二个值的类型
        JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
        	
			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			public Tuple2<String, Integer> call(String word) throws Exception {
				return new Tuple2<String, Integer>(word, 1);
			}
        	
        });
        
        // 要以单词为 key 统计每个单词出现的次数
        // 这里要使用 reduceByKey 这个算子，对每个key对应的 value 都进行 reduce 计算
        // 比如 JavaPairRDD 中分别为(hello,1)(hello,1)(hello,1)(world,1)
        // reduce 操作就是把第一个值与第二个值进行计算，然后把结果再跟第三个值进行计算
        // key 是 hello 进行 reduce 操作后，value 相当于是 1 + 1 = 2，然后 2 + 1 = 3
        // 最后返回的 JavaPairRDD 中的元素也是 truple，第一个值是key即 hello，第二个值是 value 就是3
        // 这样做 reduceByKey 的结果就是每个单词（key）出现的次数（不停的+1）
        JavaPairRDD<String, Integer> wordscount = pairs.reduceByKey(new Function2<Integer,Integer,Integer>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1 + v2;
			}
        	
        });
     
        // 到这里为止，我们通过几个 Spark 算子操作，已经统计出了单词的次数
        // flatMap, mapToPair, reduceByKey 都属于 transformation 操作,这样的操作是不会执行的，
        // 必须要有 action 的操作才能触发执行，比如 foreach
        // 
        wordscount.foreach(new VoidFunction<Tuple2<String, Integer>>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			public void call(Tuple2<String, Integer> wordcount) throws Exception {
				// TODO Auto-generated method stub
				System.out.println(wordcount._1 + "  " + wordcount._2);
			}
        	
        });
        
        // 最后要关闭 sparkContext
        sc.close();
    }
}
